%pyspark

import time
start_time = time.time()

from pyspark.sql import HiveContext
from pyspark.sql.functions import col, count, desc, sum, explode, split, to_date, year, dayofweek, lower, regexp_replace, trim
from pyspark.sql.functions import countDistinct
from pyspark.sql.types import StringType, ArrayType
from itertools import combinations
from pyspark.sql.functions import coalesce


hc = HiveContext(sc)

# 使用Hive中已存在的表
# 假设users表的字段为：user_yelping_since (而非yelping_since)
b_df = hc.table('business')
c_df = hc.table('checkin')
u_df = hc.table('users')   
r_df = hc.table('review')

##################################
# 二、用户分析
##################################

# 分析每年加入用户数 (将原先的 col("yelping_since") 改为 col("user_yelping_since"))
df12 = u_df.withColumn("year_join", year(to_date(col("user_yelping_since"), "yyyy-MM-dd"))) \
    .groupBy("year_join") \
    .agg(count("user_id").alias("new_users"))
z.show(df12)

# 评论达人（review_count最高）
df13 = u_df.select('user_id','user_review_count') \
    .orderBy(desc('user_review_count')) \
    .limit(20)
z.show(df13)

# 人气最高的用户（fans最多）
df14 = u_df.select('user_id','user_fans') \
    .orderBy(desc('user_fans')) \
    .limit(20)
z.show(df14)

# 每年优质用户（elite）与普通用户比例
# 提取用户注册年份
u_df = u_df.withColumn('register_year', year(to_date('user_yelping_since', 'yyyy-MM-dd')))

# 计算每年注册的总用户数
total_users_per_year = u_df.groupBy('register_year') \
    .agg(count('user_id').alias('total_users'))

# 提取每个用户的精英年份
elite_years = u_df.select('user_id', explode(split(col('user_elite'), ',')).alias('elite_year'))

# 过滤掉空字符串，并转换精英年份为整数
elite_years = elite_years.filter(col('elite_year') != '') \
    .withColumn('elite_year', col('elite_year').cast('int'))

# 计算每年精英用户数量
elite_users_per_year = elite_years.groupBy('elite_year') \
    .agg(countDistinct('user_id').alias('elite_users')) \
    .withColumnRenamed('elite_year', 'year')

# 将注册年份数据的列名改为 'year'
total_users_per_year = total_users_per_year.withColumnRenamed('register_year', 'year')

# 合并数据集，计算每年精英用户比例
ratio_df = total_users_per_year.join(elite_users_per_year, on='year', how='left') \
    .fillna(0) \
    .withColumn('elite_ratio', col('elite_users') / col('total_users'))

# 按年份排序并显示结果
ratio_df = ratio_df.select('year', 'total_users', 'elite_users', 'elite_ratio') \
    .orderBy('year')

z.show(ratio_df)

# 显示每年总用户数、沉默用户数比例
# 沉默用户：review_count=0 (同理，user_review_count)
# 首先获取用户加入年份列（如果之前未生成，可先生成）
u_df_year = u_df.withColumn("year_join", year(to_date(col("user_yelping_since"), "yyyy-MM-dd")))

# 每年总用户数
df16_1 = u_df_year.groupBy("year_join") \
    .agg(countDistinct("user_id").alias("total_users"))

# 每年沉默用户数
df16_2 = u_df_year.where(col("user_review_count") == 0) \
    .groupBy("year_join") \
    .agg(count("user_id").alias("silent_users"))

z.show(df16_1)
z.show(df16_2)






# 统计每年的新用户数、评论数、精英用户、打卡数 
# 统计每年的新用户数、评论数、精英用户、打卡数(不使用tip表)
year_user = u_df.withColumn("year_user", year(to_date(col("user_yelping_since"), "yyyy-MM-dd"))) \
    .groupBy("year_user").agg(countDistinct("user_id").alias("new_users"))

year_review = r_df.withColumn("year_rev", year(to_date(col("rev_date"), "yyyy-MM-dd"))) \
    .groupBy("year_rev").agg(count("*").alias("review_count"))

# 将user_elite从string分割为数组再explode
year_elite = u_df.select("user_id", explode(split(col("user_elite"), ",")).alias("elite_year")) \
    .withColumn("elite_year_int", col("elite_year").cast("int")) \
    .groupBy("elite_year_int") \
    .agg(countDistinct("user_id").alias("elite_users"))

year_ck = c_df.select(explode(split(col('checkin_dates'), ',')).alias('datetime')) \
    .withColumn('year_ck', year(to_date(trim(col('datetime')), "yyyy-MM-dd HH:mm:ss"))) \
    .groupBy('year_ck').agg(count("*").alias("checkin_count"))


result_all = year_user.join(year_review, year_user.year_user == year_review.year_rev, "full") \
    .join(year_elite, (col("year_user")==col("elite_year_int"))|(col("year_rev")==col("elite_year_int")), "full") \
    .join(year_ck, (col("year_user")==col("year_ck"))|(col("year_rev")==col("year_ck"))|(col("elite_year_int")==col("year_ck")),"full") \
    .filter(col("elite_year_int").isNotNull())

final_result = result_all.select(
    coalesce(col("year_user"), col("year_rev"), col("elite_year_int"), col("year_ck")).alias("year"),
    "new_users",
    "review_count",
    "elite_users",
    "checkin_count"
).orderBy("year")

z.show(final_result)

end_time = time.time()
execution_time = end_time - start_time
print("Execution Time: {:.2f} seconds".format(execution_time))
